package avicit.bdp.dds.server.worker.task.datax;

import avicit.bdp.common.service.dto.DataSourceDTO;
import avicit.bdp.common.utils.SpringApplicationContext;
import avicit.bdp.common.utils.database.DBUtils;
import avicit.bdp.core.datasource.enums.DataSourceType;
import avicit.bdp.dds.dispatch.enums.Flag;
import avicit.bdp.dds.common.Constants;
import avicit.bdp.dds.server.entity.TaskExecutionContext;
import avicit.bdp.dds.server.worker.task.ShellCommandExecutor;
import avicit.bdp.dds.service.process.ProcessService;
import avicit.platform6.core.exception.BusinessException;
import java.io.File;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.sql.Statement;
import java.sql.Types;
import java.util.ArrayList;
import java.util.List;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.io.FileUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;

/**
 * @author mayi DataX Increment task
 */
public class DataxIncrementTask extends DataxTask {

  private final ProcessService processService;

  /**
   * constructor
   *
   * @param taskExecutionContext taskExecutionContext
   * @param logger logger
   */
  public DataxIncrementTask(TaskExecutionContext taskExecutionContext, Logger logger) {
    super(taskExecutionContext, logger);
    this.taskExecutionContext = taskExecutionContext;
    this.shellCommandExecutor =
        new ShellCommandExecutor(this::logHandle, taskExecutionContext, logger);
    this.processService = SpringApplicationContext.getBean(ProcessService.class);
  }

  /**
   * run DataX process
   *
   * @throws Exception if error throws Exception
   */
  @Override
  public void handle() throws Exception {
    String targetTable = dataXParameters.getTargetTable();
    try {
      // 如果目标表为空，则直接执行全量同步
      DataSourceDTO targetDatasource =
          processService.findDataSourceById(dataXParameters.getDataTarget());
      Long targetCount =
          DBUtils.countTableSize(
              targetDatasource, targetTable, dataXParameters.getFilterCondition());
      if (targetCount == 0) {
        logger.info("目标表{}数据量为{},使用datax全量同步", targetTable, targetCount);
        super.handle();
      } else {
        if (StringUtils.isNotEmpty(dataXParameters.getSourceKeyColumn())
            && StringUtils.isNotEmpty(dataXParameters.getSourceIncrementColumn())
            && StringUtils.isNotEmpty(dataXParameters.getTargetKeyColumn())
            && StringUtils.isNotEmpty(dataXParameters.getTargetIncrementColumn())) {

          // 将字段按数据库类型对应转成小写或者大写
          String dstKeyColumn = dataXParameters.getTargetKeyColumn();
          String dstIncrementColumn = dataXParameters.getTargetIncrementColumn();
          dstKeyColumn =
              DBUtils.convertColumnNameToLowerCaseOrUpperCase(
                  targetDatasource.getDatasourceType(), dstKeyColumn);
          dstIncrementColumn =
              DBUtils.convertColumnNameToLowerCaseOrUpperCase(
                  targetDatasource.getDatasourceType(), dstIncrementColumn);

          dataXParameters.setIncrementSql(
              String.format(
                  "select %s %s, %s %s from %s",
                  dataXParameters.getSourceKeyColumn(),
                  dstKeyColumn,
                  dataXParameters.getSourceIncrementColumn(),
                  dstIncrementColumn,
                  dataXParameters.getSourceTable()));

          logger.info(
              "目标表{}数据量为{},使用增量量同步,目标表主键字段 = {},目标表增量字段 = {}",
              targetTable,
              targetCount,
              dstKeyColumn,
              dstIncrementColumn);

          // 1.在目标库创建临时表，用于存放源表中的主键列加增量字段列
          String tmpTable =
              "tmp_"
                  + dataXParameters.getSourceTable()
                  + Constants.UNDERLINE
                  + taskExecutionContext.getTaskInstanceId();

          // 将表名按数据库类型对应转成小写或者大写
          String convertTableName =
              DBUtils.convertTableName(targetDatasource.getDatasourceType(), tmpTable);
          tmpTable =
              DBUtils.convertTableNameToLowerCaseOrUpperCase(
                  targetDatasource.getDatasourceType(), tmpTable);

          String createTableSql =
              String.format(
                  "CREATE TABLE %s AS SELECT %s, %s FROM %s WHERE 1=2",
                  convertTableName,
                  DBUtils.convertColumnName(targetDatasource.getDatasourceType(), dstKeyColumn),
                  DBUtils.convertColumnName(
                      targetDatasource.getDatasourceType(), dstIncrementColumn),
                  dataXParameters.getTargetTable());

          if (DBUtils.isExistTable(targetDatasource, tmpTable)) {
            DBUtils.dropDataTable(targetDatasource, tmpTable);
          }
          boolean createTable = DBUtils.executeSql(targetDatasource, createTableSql);
          logger.info(
              "创建临时表{}{},创建SQL = [{}]", tmpTable, createTable ? "成功" : "失败", createTableSql);
          if (!createTable) {
            throw new BusinessException("create tmp table on target db error");
          }

          // 2.將源表的主键列和增量列同步到目标库的临时表
          String oldSql = dataXParameters.getSql();
          isSql = true;
          DATAX_CHANNEL_COUNT = 10;
          dataXParameters.setSql(dataXParameters.getIncrementSql());
          dataXParameters.setCustomConfig(Flag.NO.ordinal());
          dataXParameters.setTargetTable(tmpTable);

          try {
            super.handle();
            Long tmpTableCount =
                DBUtils.countTableSize(
                    targetDatasource, tmpTable, dataXParameters.getFilterCondition());
            logger.info("写入临时表{}完成，写入{}条", tmpTable, tmpTableCount);
            if (tmpTableCount > 0) {
              // 3.通过源表和目标库临时表查询变化的数据,并同步变化的数据
              String diffSql;
              if (DataSourceType.MySQL.getName()
                  .equalsIgnoreCase(targetDatasource.getDatasourceType())) {
                String diffSqlTemplate =
                    "SELECT targetTableName.%s targetKeyColumn,sourceTempTableName.%s sourceKeyColumn"
                        + " FROM %s targetTableName"
                        + " LEFT JOIN %s sourceTempTableName"
                        + " ON targetTableName.%s = sourceTempTableName.%s"
                        + " WHERE targetTableName.%s <> sourceTempTableName.%s"
                        + " OR sourceTempTableName.%s IS NULL"
                        + " UNION "
                        + " SELECT targetTableName.%s targetKeyColumn,sourceTempTableName.%s sourceKeyColumn"
                        + " FROM %s targetTableName"
                        + " RIGHT JOIN %s sourceTempTableName"
                        + " ON targetTableName.%s = sourceTempTableName.%s"
                        + " WHERE targetTableName.%s IS NULL ";

                diffSql =
                    String.format(
                        diffSqlTemplate,
                        dstKeyColumn,
                        dataXParameters.getSourceKeyColumn(),
                        targetTable,
                        tmpTable,
                        dstKeyColumn,
                        dataXParameters.getSourceKeyColumn(),
                        dstIncrementColumn,
                        dstIncrementColumn,
                        dstKeyColumn,
                        dataXParameters.getSourceKeyColumn(),
                        dataXParameters.getSourceKeyColumn(),
                        targetTable,
                        tmpTable,
                        dstKeyColumn,
                        dataXParameters.getSourceKeyColumn(),
                        dstKeyColumn);
              } else {
                String diffSqlTemplate =
                    "SELECT targetTableName.%s targetKeyColumn,sourceTempTableName.%s sourceKeyColumn"
                        + " FROM %s targetTableName"
                        + " FULL JOIN %s sourceTempTableName"
                        + " ON targetTableName.%s = sourceTempTableName.%s"
                        + " WHERE targetTableName.%s <> sourceTempTableName.%s"
                        + " OR targetTableName.%s IS NULL "
                        + " OR sourceTempTableName.%s IS NULL";

                diffSql =
                    String.format(
                        diffSqlTemplate,
                        dstKeyColumn,
                        dataXParameters.getSourceKeyColumn(),
                        targetTable,
                        tmpTable,
                        dstKeyColumn,
                        dataXParameters.getSourceKeyColumn(),
                        dstIncrementColumn,
                        dstIncrementColumn,
                        dstKeyColumn,
                        dataXParameters.getSourceKeyColumn());
              }

              int deleteNum = 0;
              // 需要从目标表删除的数据
              List<String> targetDeleteIdList = new ArrayList<>();
              // 需要从源表查询的数据
              List<String> sourceIdList = new ArrayList<>();

              Connection targetConnection = DBUtils.buildConnection(targetDatasource);
              Statement targetStatement = targetConnection.createStatement();
              logger.info("开始查询差异数据，sql = [{}]", diffSql);
              ResultSet rs = targetStatement.executeQuery(diffSql);
              logger.info("查询差异数据完毕");
              ResultSetMetaData metaData = rs.getMetaData();
              logger.info("开始处理增量数据>>>>>>");
              int keyColumnType = metaData.getColumnType(1);
              boolean isStringType =
                  keyColumnType == Types.CHAR
                      || keyColumnType == Types.NCHAR
                      || keyColumnType == Types.VARCHAR
                      || keyColumnType == Types.LONGVARCHAR;
              while (rs.next()) {
                String sourceKeyColumn = rs.getString("sourceKeyColumn");
                String targetKeyColumn = rs.getString("targetKeyColumn");
                // 如果目标表主键和源表临时表的主键字段均存在，表示该数据发生了变化，需要从目标表删除，从源表查询在插入目标表
                // 如果目标表主键存在，而源表临时表的主键字段不存在，表示该数据在源表已删除，则需要从目标表删除
                // 如果目标表主键不存在，而源表临时表的主键字段存在，表示该数据在源表为新增数据，则需要源表查询在插入目标表
                if (StringUtils.isNotEmpty(targetKeyColumn)
                    && StringUtils.isNotEmpty(sourceKeyColumn)) {
                  targetDeleteIdList.add(
                      isStringType ? "'" + targetKeyColumn + "'" : targetKeyColumn);
                  sourceIdList.add(isStringType ? "'" + sourceKeyColumn + "'" : sourceKeyColumn);
                } else if (StringUtils.isEmpty(targetKeyColumn)
                    && StringUtils.isNotEmpty(sourceKeyColumn)) {
                  sourceIdList.add(isStringType ? "'" + sourceKeyColumn + "'" : sourceKeyColumn);
                } else {
                  targetDeleteIdList.add(
                      isStringType ? "'" + targetKeyColumn + "'" : targetKeyColumn);
                }

                if (targetDeleteIdList.size() >= 1000) {
                  deleteNum +=
                      batchDeleteData(
                          targetConnection, targetTable, dstKeyColumn, targetDeleteIdList);
                }

                if (sourceIdList.size() >= 5000) {
                  handleChangedData(targetTable, oldSql, sourceIdList);
                }
              }

              if (CollectionUtils.isNotEmpty(targetDeleteIdList)) {
                deleteNum +=
                    batchDeleteData(
                        targetConnection, targetTable, dstKeyColumn, targetDeleteIdList);
              }

              DBUtils.closeResource(targetConnection, targetStatement, rs);
              logger.info("目标表{}删除数据{}条", targetTable, deleteNum);

              handleChangedData(targetTable, oldSql, sourceIdList);

              logger.info("datax increment task end");
            }
          } catch (Exception e) {
            throw e;
          } finally {
            DBUtils.dropDataTable(targetDatasource, tmpTable);
          }
        } else {
          throw new IllegalArgumentException("未指定增量同步字段");
        }
      }
    } catch (Exception e) {
      logger.error("datax increment task failure", e);
      setExitStatusCode(Constants.EXIT_CODE_FAILURE);
      throw e;
    }
  }

  private void handleChangedData(String targetTable, String oldSql, List<String> sourceIdList)
      throws Exception {
    if (CollectionUtils.isNotEmpty(sourceIdList)) {
      String sql = oldSql;
      if (StringUtils.isEmpty(sql)) {
        sql = "SELECT * FROM " + dataXParameters.getSourceTable();
      }
      sql += " WHERE " + getQueryIn(dataXParameters.getSourceKeyColumn(), sourceIdList);
      sourceIdList.clear();
      dataXParameters.setSql(sql);
      dataXParameters.setCustomConfig(Flag.NO.ordinal());
      dataXParameters.setTargetTable(targetTable);
      String fileName =
          String.format(
              "%s/%s_job.json",
              taskExecutionContext.getExecutePath(), taskExecutionContext.getTaskAppId());
      FileUtils.forceDelete(new File(fileName));
      super.handle();
    }
  }

  private int batchDeleteData(
      Connection connection, String tableName, String uniqueKey, List<String> list)
      throws SQLException {
    int deleteNum = 0;
    if (CollectionUtils.isNotEmpty(list)) {
      String deleteSqlTemplate = "DELETE FROM %s WHERE %s ";
      String queryIn = getQueryIn(uniqueKey, list);
      String sql = String.format(deleteSqlTemplate, tableName, queryIn);
      Statement statement = connection.createStatement();
      deleteNum = statement.executeUpdate(sql);
      list.clear();
      try {
        statement.close();
      } catch (SQLException e) {
        logger.error("close statement异常");
      }
    }
    return deleteNum;
  }

  private String getQueryIn(String filed, List<String> list) {
    StringBuilder sb = new StringBuilder();
    if (CollectionUtils.isEmpty(list)) {
      sb.append(filed).append("=''");
    } else {
      int size = list.size();
      sb.append(filed);
      sb.append(" IN (");
      for (int i = 0; i < size; i++) {
        sb.append(list.get(i));
        if (i >= 900 && i < size - 1) {
          if (i % 900 == 0) {
            sb.append(") OR ");
            sb.append(filed);
            sb.append(" IN (");
          } else {
            sb.append(Constants.COMMA);
          }
        } else {
          if (i < size - 1) {
            sb.append(Constants.COMMA);
          }
        }

        if (i == size - 1) {
          sb.append(")");
        }
      }
    }
    return sb.toString();
  }
}
